package com.gmscosmo.janus.sendMessage;

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.gmscosmo.janus.util.JanusUtil;
import com.gmscosmo.janus.queue.RabbitMqUtil;
import com.gmscosmo.model.Node;
import com.gmscosmo.request.JanusRequest;
import com.rabbitmq.client.BuiltinExchangeType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * ClassName: AsyncSender
 * Description:
 * Date: 2021/10/12 16:24
 *
 * @author stephen.qiu
 * @version 1.0
 */
@Component
public class JanusRequestSender {
    @Lazy
    @Autowired
    JanusUtil janusUtil;

    @Lazy
    @Autowired
    RabbitMqUtil rabbitMqUtil;

//    private static final CopyOnWriteArraySet<MessageObserver.Observer> OBSERVERS = new CopyOnWriteArraySet<>();

    private static final ConcurrentHashMap<String, MessageObserver.Observer> OBSERVERS = new ConcurrentHashMap<>(16);

    public interface MessageObserver {
        interface Observer {
            /**
             * 检测是否有回复
             *
             * @param transaction
             * @param data
             */
            void on(String transaction, JSONObject data);
        }
    }

    /**
     * @param transaction 回复事务号
     * @param body        信息
     * @return
     * @throws JSONException
     */
    public Future<JSONObject> sendWithFuture(String transaction, String body, int nodeId) throws JSONException {
        return new ActionCallBack(transaction, nodeId).sendAckCallback(body);
    }

    public Future<JSONObject> sendWithFuture(JanusRequest janusRequest, int nodeId) throws JSONException {
        return new ActionCallBack(janusRequest.getTransaction(), nodeId).sendAckCallback(JSONObject.toJSON(janusRequest).toString());
    }

    public static void notifyObservers(String transaction, JSONObject data) {
        try {
            MessageObserver.Observer observer = OBSERVERS.get(transaction);
            if(observer!=null) {
                observer.on(transaction, data);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private class ActionCallBack {
        private final String callbackTransaction;
        private JSONObject response;
        private final int sendNode;

        ActionCallBack(String transaction, int nodeId) {
            callbackTransaction = transaction;
            sendNode = nodeId;
        }

        public Future<JSONObject> sendAckCallback(String body) {
            CountDownLatch countDownLatch = new CountDownLatch(1);
            MessageObserver.Observer observer = (transaction, data) -> {
                if (!"ack".equals(data.getString("janus"))) {
                    response = data;
                    countDownLatch.countDown();
//                    janusUtil.removeTransaction(callbackTransaction);
                }
            };
            OBSERVERS.put(callbackTransaction, observer);
            Node node = janusUtil.getNodeInfo(sendNode);
            rabbitMqUtil.sendMq(node.getTo_janus(), node.getTo_janus(), BuiltinExchangeType.DIRECT, body);
            janusUtil.addTransaction(callbackTransaction);
            try {
                countDownLatch.await(5, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDownLatch.countDown();
            }
            OBSERVERS.remove(observer);
            return new AsyncResult<>(response);
        }
    }
}
